package time_consume;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import workCount.HDFS_client;

import java.io.IOException;
import java.net.URISyntaxException;

public class MainDriver {
    public static void main(String[] args) throws IOException, URISyntaxException, InterruptedException, ClassNotFoundException {
        //自定义变量
        //定义HDFS链接地址，需要自行修改
        String URL = "hdfs://192.168.152.128:9000";
        //定义数据所在的windows本地路径，需要自行修改wordCount文件存放位置
        //操作文件名字为wordCount.txt，路径自定义
        //在Windows下的路径表达
        String localDir = "D:\\新建文件夹\\data\\files\\time_consume.log";
        //在Linux下的路径表达
        //String localDir = "/root/wordCount.txt";

        //固定变量
        //定义HDFS用户
        String user = "root";
        //定义处理的原始数据导入路径
        String srcDir = "/input";
        //定义处理完成输出的路径
        String destDir = "/output";
        //需要处理的文件所在的HDFS位置
        String fileName = srcDir + "/time_consume.log";

        HDFS_client client = new HDFS_client(URL, user);
        //判定目录是否存在，如果不存在则创建目录，如果存在则删除原有目录并创建
        client.createDir(srcDir);

        //上传需要操作的wordCount文件
        client.uploadFile(localDir, srcDir);

        //查看上传的文件内容
        System.out.println("==========\n" + "当前需要处理的文件内容");
        client.printFile(fileName);

        //删除输出的目录
        client.deleteFile(destDir);

        //执行MRDemo内容
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", URL);
        Job job = Job.getInstance(conf, "MRDemo");

        //定义MR任务的文件输入输出来源
        FileInputFormat.addInputPath(job, new Path(fileName));
        FileOutputFormat.setOutputPath(job, new Path(destDir));

        //设置job的主任务、map任务、reduce任务
        job.setJarByClass(MainDriver.class);
        job.setMapperClass(MapperDemo.class);
        job.setReducerClass(ReducerDemo.class);

        //Map的输出类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(BeanDemo.class);

        //Reduce输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(BeanDemo.class);

        //执行MR任务
        job.waitForCompletion(true);

        //查看处理完成的文件内容
        System.out.println("==========" + "\n当前处理完成的文件内容\n" + "==========");
        client.printFile(destDir);

        //断开与HDFS的连接
        client.close();
    }
}
